package com.liuqi.iot.collector;

import com.liuqi.iot.bean.GatewayConnectionInfo;
import com.liuqi.iot.bean.dto.GatewayDTO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Service;

import java.util.Hashtable;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * MQTT的实时工况直接处理器
 *
 * @author  LiuQi 2025/3/13-9:21
 * @version V1.0
 **/
@Service
@Slf4j
public class MqttDataCollector extends AbstractDataCollector {
    private static final String CLIENT_ID = "qi-iot";

    private static final Map<String, MqttClient> clientMap = new Hashtable<>();

    private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10);

    // 停止监听
    @Override
    public void stopListener(String gatewayId) {
        Optional.ofNullable(clientMap.get(gatewayId)).ifPresent(client -> {
            try {
                if (!client.isConnected()) {
                    clientMap.remove(gatewayId);
                    return;
                }
                client.disconnect();
                clientMap.remove(gatewayId);
            } catch (MqttException e) {
                log.error("关闭连接失败", e);
            }
        });
    }

    /**
     * 客户端重连
     */
    private void reconnect(GatewayDTO gateway, MqttClient client) {
        scheduler.schedule(() -> {
            try {
                client.connect();
                client.subscribe(gateway.getTopic());
            } catch (Exception ex) {
                log.error("重连失败", ex);
                reconnect(gateway, client);
            }
        }, 60, TimeUnit.SECONDS);
    }

    /**
     * 启动MQTT监听
     */
    @Override
    public void startListener(GatewayDTO gateway) throws MqttException {
        // 需要判断之前的连接是否存在
        String gatewayId = gateway.getId();
        this.stopListener(gatewayId);

        GatewayConnectionInfo connectionInfo = gateway.getConnectionInfo();
        MqttClient client = new MqttClient("tcp://" + connectionInfo.getHost() + ":" + connectionInfo.getPort(),
                CLIENT_ID + "_" + gatewayId, new MemoryPersistence());
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                // 手动关闭时不会进这里
                reconnect(gateway, client);
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                // 处理消息
                String msg = message.toString();
                try {
                    log.debug("开始处理mqtt消息：{}", gateway.getName());
                    process(gateway, topic, msg);
                    log.debug("MQTT消息处理完成：{}", gateway.getName());
                } catch (Exception ex) {
                    log.error("mqtt消息处理失败", ex);
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {

            }
        });

        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setCleanSession(true);
        connOpts.setConnectionTimeout(10);
        connOpts.setAutomaticReconnect(false);
        client.connect(connOpts);
        clientMap.put(gatewayId, client);

        // 订阅主题
        client.subscribe(gateway.getTopic());
    }
}
